package main

import (
	"context"

	"task/config"
	"task/driver"
	service "task/service/monitor"

	"github.com/apache/pulsar-client-go/pulsar"
	log "github.com/sirupsen/logrus"
)

const channelLen = 100000

func main() {
	var (
		consumer, callbackConsumer pulsar.Consumer
		err                        error
	)
	// 初始化消息队列
	driver.InitPulsarClient()
	defer driver.PulsarClient.Close()
	// 开启metrics服务
	go driver.StartMetricsServer()
	// 初始化dispatch grpc客户端
	conn := driver.InitDispatchClient()
	defer conn.Close()
	// 创建调度消息消费者
	channel := make(chan pulsar.ConsumerMessage, channelLen)
	options := pulsar.ConsumerOptions{
		Topic:            config.Conf.Pulsar.DispatchTopic,
		SubscriptionName: config.Conf.Pulsar.SubscriptionName,
		Type:             pulsar.Shared,
	}
	options.MessageChannel = channel
	if consumer, err = driver.PulsarClient.Subscribe(options); err != nil {
		log.Panic("Failed to create pulsar consumer:", err)
	}
	defer consumer.Close()
	// 创建回调消息消费者
	callbackChannel := make(chan pulsar.ConsumerMessage, channelLen)
	callbackOptions := pulsar.ConsumerOptions{
		Topic:            config.Conf.Pulsar.CallbackTopic,
		SubscriptionName: config.Conf.Pulsar.SubscriptionName,
		Type:             pulsar.Shared,
	}
	callbackOptions.MessageChannel = callbackChannel
	if callbackConsumer, err = driver.PulsarClient.Subscribe(callbackOptions); err != nil {
		log.Panic("Failed to create pulsar consumer:", err)
	}
	defer callbackConsumer.Close()
	// 监听调度事件消息
	go func() {
		log.Info("Start to monitor dispatch message...")
		for cm := range channel {
			msg := cm.Message
			log.Infof("Received dispatch message: %s channel len: %d", string(msg.Payload()), len(channel))
			service.SetConsumerChLen(float64(len(channel)))
			_ = service.StartDispatch(context.Background(), msg.Payload())
			// 无论成功失败 都要消息确认
			consumer.Ack(msg)
		}
	}()
	// 监听回调消息
	go func() {
		log.Info("Start to monitor callback message...")
		for cm := range callbackChannel {
			msg := cm.Message
			log.Infof("Received callback message: %s channel len: %d", string(msg.Payload()), len(channel))
			// 回调消息去调度
			_ = service.StartCallback(context.Background(), msg.Payload())
			// 无论成功失败 都要消息确认
			consumer.Ack(msg)
		}
	}()
	// 阻塞main
	select {}
}
